package org.teemor.demo.kafka.springclient.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Iterator;
import java.util.List;

/**
 * @Description:
 * @author zhoulk
 * @date: 2020/2/13
 */
@Component
public class SimpleConsumer {

//    @KafkaListener(containerFactory = "simpleFactory", topics = "test1")
//    public void simpleConsume(String content) {
//        System.out.println(content);
//    }

    @KafkaListener(containerFactory = "simpleFactory", topics = "test")
    public void ackConsume(List<ConsumerRecord> consumerRecords, Acknowledgment ack) {
        ack.acknowledge();
        if (consumerRecords.size() > 0) {
            System.out.println(consumerRecords.get(0).partition() + ":" + consumerRecords.size());
        }
        Iterator<ConsumerRecord> iterator = consumerRecords.iterator();
        while (iterator.hasNext()) {
            ConsumerRecord consumerRecord = iterator.next();
            String key = consumerRecord.key().toString();
            System.out.println("key:" + key + " || value:" + consumerRecord.value());
        }
    }

    @KafkaListener(topics = "test2")
    public void defaultConsume(String content) {
        System.out.println(content);
    }
}
